package com.aispeech.asr.webase.seaweedfs.client.impl;

import com.aispeech.asr.webase.seaweedfs.client.Assignation;
import com.aispeech.asr.webase.seaweedfs.client.Location;
import com.aispeech.asr.webase.seaweedfs.client.SeaWeedFSClient;
import com.aispeech.asr.webase.seaweedfs.client.WeedFSFile;
import com.aispeech.asr.webase.seaweedfs.client.cache.LookupCache;
import com.aispeech.asr.webase.seaweedfs.client.exceptions.WeedFSException;
import com.aispeech.asr.webase.seaweedfs.client.request.AssignParams;
import com.aispeech.asr.webase.seaweedfs.client.request.Manifest;
import com.aispeech.asr.webase.seaweedfs.client.response.result.AssignResult;
import com.aispeech.asr.webase.seaweedfs.client.response.status.MasterStatus;
import com.aispeech.asr.webase.seaweedfs.client.response.status.VolumeStatus;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.*;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.MediaType;
import org.springframework.http.client.MultipartBodyBuilder;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.File;
import java.net.URI;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class SeaWeedFSClientImpl implements SeaWeedFSClient {
    WebClient webClient;
    LookupCache lookupCache;
    String masterUrl;
    ObjectMapper objectMapper;

    public SeaWeedFSClientImpl(WebClient webClient, String masterUrl, LookupCache lookupCache) {
        this.webClient = webClient;
        this.lookupCache = lookupCache;
        this.masterUrl = masterUrl;

        objectMapper = new ObjectMapper();
    }

    @Override
    public Mono<Assignation> assign() {
        return assign(AssignParams.DEFAULT);
    }

    @Override
    public Mono<Assignation> assign(AssignParams assignParams) {
        URI uri = URI.create(masterUrl + "/dir/assign");

        Map<String, String> params = new HashMap<>();
        params.put("count", String.valueOf(assignParams.versionCount));
        if (assignParams.replicationStrategy != null) {
            params.put("replication", assignParams.replicationStrategy.parameterValue);
        }
        if (assignParams.collection != null) {
            params.put("collection", assignParams.collection);
        }

        return webClient.get()
                .uri(masterUrl + "/dir/assign", params)
                .retrieve()
                .bodyToMono(AssignResult.class)
                .map(assignResult -> new Assignation(assignResult));
    }

    public Mono<Integer> write(Assignation assignation, File file) {
        if (file.length() == 0) {
            return Mono.error(new WeedFSException("Cannot write a 0-length file"));
        } else {
            return write(assignation, new FileSystemResource(file.getAbsolutePath()), file.length());
        }
    }

    private Mono<Integer> write(Assignation assignation, AbstractResource resource, long resourceLength) {
        StringBuilder url = new StringBuilder();
        if (!assignation.location.publicUrl.contains("http")) {
            url.append("http://");
        }
        url.append(assignation.location.publicUrl);
        url.append('/');
        url.append(assignation.weedFSFile.fid);

        if (assignation.weedFSFile.version > 0) {
            url.append('_');
            url.append(assignation.weedFSFile.version);
        }

        MultipartBodyBuilder builder = new MultipartBodyBuilder();
        builder.part("file", resource);

//        log.info("[ write ] url: {}", url.toString());

        return webClient.post()
                .uri(url.toString())
                .contentType(MediaType.MULTIPART_FORM_DATA)
                .body(BodyInserters.fromMultipartData(builder.build()))
                .retrieve()
                .bodyToMono(JsonNode.class)
                .flatMap(jsonNode -> {
                    int length = jsonNode.get("size").asInt();

                    if (length == resourceLength)
                        return Mono.just(length);
                    else
                        return Mono.error(new WeedFSException(
                                String.format("upload failed. length: %d, uploaded: %d",
                                        resourceLength, length))
                        );

                });
    }

    public Mono<Integer> write(Assignation assignation, DataBuffer buffer) {
        if (buffer.readableByteCount() == 0) {
            return Mono.error(new WeedFSException("Cannot write a 0-length file"));
        } else {
            return write(assignation, new InputStreamResource(buffer.asInputStream()), buffer.readableByteCount());
        }
    }

    public Mono<Integer> write(Assignation assignation, String text) {
        if (text.length() == 0) {
            return Mono.error(new WeedFSException("Cannot write a 0-length text"));
        } else {
            return write(assignation, new ByteArrayResource(text.getBytes()), text.length());
        }
    }

    public Mono<Assignation> merge(Manifest manifest) {
        byte[] bytes;

        try {
            bytes = objectMapper.writeValueAsBytes(manifest);
        } catch (JsonProcessingException e) {
            return Mono.error(new WeedFSException(e.getMessage()));
        }

        return assign()
                .flatMap(assignation -> {
                    StringBuilder url = new StringBuilder();

                    if (!assignation.location.publicUrl.contains("http")) {
                        url.append("http://");
                    }
                    url.append(assignation.location.publicUrl);
                    url.append('/');
                    url.append(assignation.weedFSFile.fid);

                    if (assignation.weedFSFile.version > 0) {
                        url.append('_');
                        url.append(assignation.weedFSFile.version);
                    }
                    url.append("?cm=true");

                    MultipartBodyBuilder builder = new MultipartBodyBuilder();
                    builder.part("file", new ByteArrayResource(bytes), MediaType.APPLICATION_JSON);

//                    log.info("[ write ] url: {}", url.toString());

                    return webClient.post()
                            .uri(url.toString())
                            .contentType(MediaType.APPLICATION_JSON)
                            .body(BodyInserters.fromMultipartData(builder.build()))
                            .retrieve()
                            .bodyToMono(JsonNode.class)
                            .map(jsonNode -> assignation);
                });
    }

    @Override
    public Mono<Void> delete(WeedFSFile file, Location location) {
        StringBuilder url = new StringBuilder();
        if (!location.publicUrl.contains("http")) {
            url.append("http://");
        }
        url.append(location.publicUrl);
        url.append("/");
        url.append(file.fid);

        return webClient.delete()
                .uri(url.toString())
                .retrieve()
                .bodyToMono(Void.class);
    }

    @Override
    public Flux<Location> lookup(long volumeId) {
        List<Location> ret = lookupCache.lookup(volumeId);

        if (ret != null) {
            return Flux.fromIterable(ret);
        } else {
            return webClient.get()
                    .uri(masterUrl + "/dir/lookup", new HashMap<String, Long>() {{
                        put("volumeId", volumeId);
                    }})
                    .retrieve()
                    .bodyToFlux(Location.class);
        }
    }

    @Override
    public Mono<Resource> read(WeedFSFile file, Location location) {
        StringBuilder url = new StringBuilder();
        if (!location.publicUrl.contains("http")) {
            url.append("http://");
        }
        url.append(location.publicUrl);
        url.append('/');
        url.append(file.fid);

        if (file.version > 0) {
            url.append('_');
            url.append(file.version);
        }

        return webClient.get()
                .uri(url.toString())
                .retrieve()
                .bodyToMono(Resource.class);
    }

    @Override
    public Mono<MasterStatus> getMasterStatus() {
        return webClient.get()
                .uri(masterUrl + "/dir/status")
                .retrieve()
                .bodyToMono(MasterStatus.class);
    }

    @Override
    public Mono<VolumeStatus> getVolumeStatus(Location location) {
        StringBuilder url = new StringBuilder();
        if (!location.publicUrl.contains("http")) {
            url.append("http://");
        }
        url.append(location.publicUrl);
        url.append("/status");

        return webClient.get()
                .uri(masterUrl + "/dir/status")
                .retrieve()
                .bodyToMono(VolumeStatus.class);
    }
}
